sync job works concurrently #190

SrinivasChilveri · 2019-05-20T06:09:11Z

sync jobs concurrently

Fix : #93

SrinivasChilveri · 2019-05-20T07:29:38Z

@hzxuzhonghu , pls do have a look.

lminzhw · 2019-05-20T09:25:07Z

pkg/controllers/job/job_controller.go

-	go wait.Until(cc.worker, 0, stopCh)
+
+	for i := 0; i < workers; i++ {
+		go wait.Until(cc.worker, time.Second, stopCh)


Why we use time.Second here but not 0 like the old logic? ...

@lminzhw followed k8s controller mgr controllers way.I hope if we have many controllers then its good to have some time instead of zero. may be we can have this parameter also as config parameter. what do you say?

I browse some code of controller in k8s, and realize that most of k8s controller use time.Second as interval and multi-worker to solve the large scale problem.

Seems good :)

hzxuzhonghu · 2019-05-29T10:43:32Z

@SrinivasChilveri it is not that simple to concurrently run multi workers, the workers may conflict operating same Job i think. Have to prevent this.

SrinivasChilveri · 2019-05-30T09:52:55Z

@SrinivasChilveri it is not that simple to concurrently run multi workers, the workers may conflict operating same Job i think. Have to prevent this.

yeah, as our controller can have multiple entries related to same job in queue. have pushed the commit. pls do have a look

SrinivasChilveri · 2019-05-31T06:05:33Z

@hzxuzhonghu ,added new commit for your comment. pls do have a look.

k82cn · 2019-06-04T09:35:41Z

pkg/controllers/job/job_controller.go

+	cc.jobsLock.Lock()
+	if cc.jobsMap[key] {
+		// the job is being processed by some other thread
+		cc.queue.AddRateLimited(req)


if conflict after a few times, it'll take a really time to retry job. It's better to shard the request to different chanel/worker.

Also this re-enqueue ops can lead to request processed in unexpected orders, though i have no detailed scenario now.

@k82cn , @hzxuzhonghu , Based on my understading sharding with single queue will not solve this issue. so i have added queueList with number of workers & did the commit. please do review the same.

k82cn · 2019-06-17T08:40:45Z

pkg/controllers/job/job_controller.go

+	// Later we can remove this code if we want
+	if !cc.belongsToThisRoutine(key, count) {
+		glog.Errorf("should not occur The job does not belongs to this routine key:%s, worker:%d...... ", key, count)
+		cc.queueList[count].AddRateLimited(req)


do not use AddRateLimited in this case; just hash request by job into different queue, and have several workers to handle them.

hzxuzhonghu

some nits

hzxuzhonghu · 2019-06-17T09:10:44Z

pkg/controllers/job/job_controller.go

-func (cc *Controller) processNextReq() bool {
-	obj, shutdown := cc.queue.Get()
+func (cc *Controller) processNextReq(count uint32) bool {
+	obj, shutdown := cc.queueList[count].Get()


nit: I would like to acquire the queue before this loop

hzxuzhonghu · 2019-06-17T09:12:11Z

pkg/controllers/job/job_controller.go

+	defer cc.queueList[count].Done(req)
+
+	key := jobcache.JobKeyByReq(&req)
+	// Later we can remove this code if we want


why add this?

Basically we are sharing the jobs and adding into different queues. In this function just reading from different queues ,so it never happens to hit success this condition because the while placing the job request into different queue based on shard. while reading also it should belong to same shard / queue/worker.

let's remove that

@SrinivasChilveri

hzxuzhonghu · 2019-06-17T09:13:28Z

pkg/controllers/job/job_controller.go

+	}
+
+	// prevent multi threads processing the same job simultaneously.
+	cc.jobsLock.Lock()


This would not happen if you do sharding in the source.

agreed, will remove the code

hzxuzhonghu · 2019-06-17T09:15:17Z

pkg/controllers/job/job_controller_handler.go

@@ -241,7 +247,9 @@ func (cc *Controller) updatePod(oldObj, newObj interface{}) {
 		JobVersion: int32(dVersion),
 	}

-	cc.queue.Add(req)
+	key := vkcache.JobKeyByReq(&req)
+	i := cc.getWorkerID(key)


I think this can guarantee worker threads will not process same jobs

hzxuzhonghu · 2019-06-17T09:17:59Z

I'd like to see some tests on concurrent accessing the job cache

SrinivasChilveri · 2019-06-18T07:36:18Z

I'd like to see some tests on concurrent accessing the job cache

better to do in separate PR

hzxuzhonghu · 2019-06-26T09:20:52Z

LGTM, leaves to @k82cn for last eye

SrinivasChilveri · 2019-06-26T09:21:28Z

@k82cn @hzxuzhonghu , pls do have a look and let me know, need to update any more points.

SrinivasChilveri · 2019-06-27T08:43:44Z

@k82cn , the PR rebased.

SrinivasChilveri · 2019-07-03T12:20:53Z

@k82cn do i need to change anything else.

TommyLike · 2019-07-08T09:33:06Z

pkg/controllers/job/job_controller.go

+}
+
+// TODO we may need to make this sharding more proper if required
+func (cc *Controller) getWorkerID(key string) workqueue.RateLimitingInterface {


This function doesn't work as its name describes

changed the name

TommyLike · 2019-07-09T06:10:41Z

/lgtm

TommyLike · 2019-07-09T06:12:15Z

cc @k82cn @hzxuzhonghu

hzxuzhonghu · 2019-07-09T06:23:44Z

/lgtm

k82cn · 2019-07-10T01:32:23Z

cmd/controllers/app/server.go

 	queueController := queue.NewQueueController(kubeClient, kbClient)
 	garbageCollector := garbagecollector.New(vkClient)

 	run := func(ctx context.Context) {
-		go jobController.Run(ctx.Done())
+		go jobController.Run(opt.WorkerThreads, ctx.Done())


We already create jobController with opt.WorkerThreads, why Run still need this parameters?

k82cn · 2019-07-10T01:32:38Z

pkg/controllers/job/job_controller.go

@@ -21,6 +21,9 @@ import (
 	"sync"

 	"github.com/golang/glog"
+	"hash"


close to other system libs.

k82cn · 2019-07-10T01:33:03Z

pkg/controllers/job/job_controller.go

 }

 // NewJobController create new Job Controller
 func NewJobController(
 	kubeClient kubernetes.Interface,
 	kbClient kbver.Interface,
 	vkClient vkver.Interface,
+	workers uint32,
+


remove empty line

k82cn · 2019-07-10T01:34:37Z

pkg/controllers/job/job_controller.go

-func (cc *Controller) worker() {
-	for cc.processNextReq() {
+func (cc *Controller) worker(i uint32) {
+


remove empty line.

k82cn · 2019-07-10T01:34:45Z

pkg/controllers/job/job_controller.go

+				time.Second,
+				stopCh)
+		}(i)
+


remove empty line.

k82cn · 2019-07-10T01:38:04Z

pkg/controllers/job/job_controller_handler.go

@@ -65,7 +65,9 @@ func (cc *Controller) addJob(obj interface{}) {
 		glog.Errorf("Failed to add job <%s/%s>: %v in cache",
 			job.Namespace, job.Name, err)
 	}
-	cc.queue.Add(req)
+	key := vkcache.JobKeyByReq(&req)


introduces a helper function for that.

pkg/controllers/job/job_controller_plugins_test.go

SrinivasChilveri · 2019-07-10T07:11:22Z

@hzxuzhonghu @k82cn @TommyLike , Fixed all the comments

pkg/controllers/job/job_controller.go

hzxuzhonghu · 2019-07-10T11:32:45Z

/lgtm

hzxuzhonghu · 2019-07-10T11:33:10Z

/approve

SrinivasChilveri · 2019-07-11T06:36:18Z

@TommyLike @hex108 any other suggestions.

hzxuzhonghu · 2019-07-11T06:39:37Z

ping @k82cn for approval

TommyLike · 2019-07-11T06:44:38Z

/lgtm

k82cn · 2019-07-12T01:26:20Z

/approve

volcano-sh-bot · 2019-07-12T01:26:36Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: hzxuzhonghu, k82cn, SrinivasChilveri

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~OWNERS~~ [k82cn]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

lminzhw reviewed May 20, 2019

View reviewed changes

k82cn reviewed Jun 4, 2019

View reviewed changes

SrinivasChilveri force-pushed the SynchJobConcurrently branch from 28cab1c to c00d12c Compare June 17, 2019 07:16

k82cn reviewed Jun 17, 2019

View reviewed changes

hzxuzhonghu reviewed Jun 17, 2019

View reviewed changes

volcano-sh-bot added the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Jun 27, 2019

SrinivasChilveri added 4 commits June 27, 2019 12:03

sync job works concurrently

07471c3

preventing the same job handling by multiple workers

7521589

added sharding with multiple queues

6a925b9

Fixed comments

acf0193

SrinivasChilveri force-pushed the SynchJobConcurrently branch from f33c5e6 to acf0193 Compare June 27, 2019 06:59

volcano-sh-bot added the size/L Denotes a PR that changes 100-499 lines, ignoring generated files. label Jun 27, 2019

volcano-sh-bot removed the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Jun 27, 2019

TommyLike reviewed Jul 8, 2019

View reviewed changes

volcano-sh-bot assigned TommyLike Jul 9, 2019

volcano-sh-bot added the lgtm Indicates that a PR is ready to be merged. label Jul 9, 2019

volcano-sh-bot assigned hzxuzhonghu Jul 9, 2019

k82cn reviewed Jul 10, 2019

View reviewed changes

pkg/controllers/job/job_controller_plugins_test.go Outdated Show resolved Hide resolved

volcano-sh-bot removed the lgtm Indicates that a PR is ready to be merged. label Jul 10, 2019

hzxuzhonghu reviewed Jul 10, 2019

View reviewed changes

pkg/controllers/job/job_controller.go Show resolved Hide resolved

Fixed all the comments

eee74aa

SrinivasChilveri force-pushed the SynchJobConcurrently branch from bff14cf to eee74aa Compare July 10, 2019 11:10

volcano-sh-bot added the lgtm Indicates that a PR is ready to be merged. label Jul 10, 2019

volcano-sh-bot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Jul 12, 2019

volcano-sh-bot merged commit c546f9f into volcano-sh:master Jul 12, 2019

sync job works concurrently #190

sync job works concurrently #190

Conversation

SrinivasChilveri commented May 20, 2019

SrinivasChilveri commented May 20, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hzxuzhonghu commented May 29, 2019

SrinivasChilveri commented May 30, 2019

SrinivasChilveri commented May 31, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hzxuzhonghu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hzxuzhonghu commented Jun 17, 2019

SrinivasChilveri commented Jun 18, 2019 • edited Loading

hzxuzhonghu commented Jun 26, 2019

SrinivasChilveri commented Jun 26, 2019

SrinivasChilveri commented Jun 27, 2019

SrinivasChilveri commented Jul 3, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TommyLike commented Jul 9, 2019

TommyLike commented Jul 9, 2019

hzxuzhonghu commented Jul 9, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SrinivasChilveri commented Jul 10, 2019

hzxuzhonghu commented Jul 10, 2019

hzxuzhonghu commented Jul 10, 2019

SrinivasChilveri commented Jul 11, 2019

hzxuzhonghu commented Jul 11, 2019

TommyLike commented Jul 11, 2019

k82cn commented Jul 12, 2019

volcano-sh-bot commented Jul 12, 2019

SrinivasChilveri commented Jun 18, 2019 •

edited

Loading